package com.swak.job.manager;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import org.apache.dubbo.config.annotation.DubboService;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import com.google.common.collect.Lists;
import com.swak.Constants;
import com.swak.OS;
import com.swak.annotation.FluxService;
import com.swak.entity.Page;
import com.swak.entity.Parameters;
import com.swak.exception.StaleObjectStateException;
import com.swak.job.JobEvent;
import com.swak.job.JobManager;
import com.swak.job.JobService;
import com.swak.job.domain.JobDTO;
import com.swak.job.domain.JobDispatchDTO;
import com.swak.job.domain.JobEventDTO;
import com.swak.job.domain.JobFailDTO;
import com.swak.job.domain.JobTaskDTO;
import com.swak.job.storage.StorageService;
import com.swak.persistence.OrderBy;
import com.swak.persistence.QueryCondition;
import com.swak.registry.NotifyListener;
import com.swak.registry.RegistryService;
import com.swak.registry.URL;
import com.swak.utils.Maps;
import com.swak.utils.Sets;
import com.swak.utils.StringUtils;
import com.swak.utils.time.DateTimes;

/**
 * 实现 Job 管理服务
 * 
 * @author 618lf
 */
@FluxService
@DubboService
public class JobManagerImpl implements JobManager, InitializingBean, NotifyListener {

	@Autowired
	private JobService jobService;
	@Autowired
	private StorageService storageService;
	@Autowired
	private RegistryService registryService;
	private volatile Map<String, Set<URL>> jobs = Maps.newHashMap();

	@Override
	public void afterPropertiesSet() throws Exception {
		Map<String, String> parameters = Maps.newHashMap();
		parameters.put(Constants.Application_KEY, registryService.applicationName());
		parameters.put(Constants.Module_Key, jobService.getConfig().getModule());
		parameters.put(Constants.Service_Key, Constants.Subscribe_Any_Value);
		parameters.put(Constants.Category_Key,
				Constants.Registry_Job_Executer_Key + Constants.IDS_SEPARATE + Constants.Registry_Job_Dispatcher_Key);
		URL url = new URL(Constants.Registry_Job_Key, OS.ip(), 0, Constants.Registry_Job_Worker_Key, parameters);
		this.registryService.subscribe(url, this);
	}

	@Override
	public void notify(List<URL> urls) {
		for (URL url : urls) {
			String service = url.getService();
			Set<URL> services = this.jobs.computeIfAbsent(service, (k) -> {
				return Sets.newHashSet();
			});
			services.add(url);
		}
	}

	@Override
	public List<URL> queryDispatchers(String jobId) {
		Set<URL> jobs = this.jobs.get(jobId);
		if (!Sets.isEmpty(jobs)) {
			return jobs.stream().filter(url -> Constants.Registry_Job_Dispatcher_Key.equals(url.getCategory()))
					.collect(Collectors.toList());
		}
		return Lists.newArrayList();
	}

	@Override
	public List<URL> queryExecuters(String jobId) {
		Set<URL> jobs = this.jobs.get(jobId);
		if (!Sets.isEmpty(jobs)) {
			return jobs.stream().filter(url -> Constants.Registry_Job_Executer_Key.equals(url.getCategory()))
					.collect(Collectors.toList());
		}
		return Lists.newArrayList();
	}

	@Override
	public Boolean dispatch(JobEventDTO event) {
		return this.dispatch(new JobEvent().setJobId(event.getJobId()).setConfig(event.getConfig())
				.setDispatchTime(event.getDispatchTime()));
	}

	@Override
	public Boolean dispatch(String jobId) {
		return this.dispatch(jobId, StringUtils.EMPTY);
	}

	@Override
	public Boolean dispatch(String jobId, String config) {
		return this.dispatch(jobId, config, DateTimes.now());
	}

	@Override
	public Boolean dispatch(String jobId, String config, LocalDateTime delayTime) {
		return this.dispatch(new JobEvent().setJobId(jobId).setConfig(config).setDispatchTime(delayTime));
	}

	private Boolean dispatch(JobEvent event) {

		// 校验任务是否存在
		if (event == null || StringUtils.isBlank(event.getJobId())) {
			throw new StaleObjectStateException("需要指定运行的任务！");
		}

		// 如果没有指定则立即执行
		if (event.getDispatchTime() == null) {
			event.setDispatchTime(DateTimes.now());
		}

		// 创建一个事件任务
		this.storageService.createEvent(event);
		return true;
	}

	@Override
	public Page<JobDTO> jobPage(JobDTO query, Parameters param) {
		QueryCondition condition = new QueryCondition();
		if (StringUtils.isNotBlank(query.getName())) {
			condition.getCriteria().andLike("NAME", query.getName());
		}
		return storageService.jobPage(condition, param);
	}

	@Override
	public Page<JobDispatchDTO> dispatchPage(JobDispatchDTO query, Parameters param) {
		QueryCondition qc = new QueryCondition();
		if (query.getJobId() != null) {
			qc.getCriteria().andEqualTo("JOB_ID", query.getJobId());
		}
		qc.setOrderByClause(OrderBy.of().Desc("CREATE_TIME"));
		return storageService.dispatchPage(qc, param);
	}

	@Override
	public Page<JobTaskDTO> taskPage(JobTaskDTO query, Parameters param) {
		QueryCondition qc = new QueryCondition();
		if (query.getJobId() != null) {
			qc.getCriteria().andEqualTo("JOB_ID", query.getJobId());
		}
		qc.setOrderByClause(OrderBy.of().Desc("CREATE_TIME"));
		return storageService.taskPage(qc, param);
	}

	@Override
	public Page<JobFailDTO> failPage(JobFailDTO query, Parameters param) {
		QueryCondition qc = new QueryCondition();
		if (query.getJobId() != null) {
			qc.getCriteria().andEqualTo("JOB_ID", query.getJobId());
		}
		qc.setOrderByClause(OrderBy.of().Desc("CREATE_TIME"));
		return storageService.failPage(qc, param);
	}

	@Override
	public Page<JobEventDTO> eventPage(JobEventDTO query, Parameters param) {
		QueryCondition condition = new QueryCondition();
		return storageService.eventPage(condition, param);
	}
}
